import os
import pytest
from unittest.mock import patch
from motor.utils.http_client import SafeHTTPSClient
import requests
import copy
from urllib.parse import urlparse

from motor.resources.instance import Instance, PDRole, Endpoint
from motor.coordinator.core.instance_manager import InstanceManager
from motor.coordinator.metrics.metrics_listener import MetricsListener
from motor.coordinator.metrics.metrics import Metrics


def mocked_requests_get(*args, **kwargs):
    class MockResponse:
        def __init__(self, json_data, status_code):
            self.json_data = json_data
            self.status_code = status_code

        def json(self):
            return self.json_data

class MockResponse:
    def __init__(self, json_data, status_code):
        self.json_data = json_data
        self.status_code = status_code

    def json(self):
        return self.json_data

class TestMetrics:
    def setup_method(self):
        ep0 = Endpoint(id=0, ip="127.0.0.1", port="8000")
        ep1 = Endpoint(id=1, ip="127.0.0.1", port="8001")
        ep2 = Endpoint(id=2, ip="127.0.0.1", port="8002")
        ep3 = Endpoint(id=3, ip="127.0.0.1", port="8003")
        ep4 = Endpoint(id=4, ip="127.0.0.1", port="8004")
        ep5 = Endpoint(id=5, ip="127.0.0.1", port="8005")
        p_ins = Instance(
            job_name="test-prefill",
            model_name="test-model",
            id=0,
            role=PDRole.ROLE_P,
            endpoints={
                "127.0.0.1": { 0: ep0, 1: ep1 }
            }
        )
        d_ins = Instance(
            job_name="test-decode",
            model_name="test-model",
            id=1,
            role=PDRole.ROLE_D,
            endpoints={
                "127.0.0.1": { 2: ep2, 3: ep3 }
            }
        )
        h_ins = Instance(
            job_name="test-hybrid",
            model_name="test-model",
            id=2,
            role=PDRole.ROLE_U,
            endpoints={
                "127.0.0.1": { 4: ep4, 5: ep5 }
            }
        )

        self.all_ins = {
            PDRole.ROLE_P: {
                p_ins.id: p_ins
            },
            PDRole.ROLE_D: {
                d_ins.id: d_ins
            },
            PDRole.ROLE_U: {
                h_ins.id: h_ins
            }
        }

        self.metrics_template = """# HELP vllm:num_requests_running Number of requests in model execution batches.
# TYPE vllm:num_requests_running gauge
vllm:num_requests_running{engine="0",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 1.0
# HELP vllm:gpu_prefix_cache_queries_total GPU prefix cache queries, in terms of number of queriedtokens. DEPRECATED: Use vllm:prefix_cache_queries instead.
# TYPE vllm:gpu_prefix_cache_queries_total counter
vllm:gpu_prefix_cache_queries_total{engine="0",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 39.0
# HELP vllm:request_success_total Count of successfully processed requests.
# TYPE vllm:request_success_total counter
vllm:request_success_total{engine="0",finished_reason="stop",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 1.0
vllm:request_success_total{engine="0",finished_reason="length",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 2.0
vllm:request_success_total{engine="0",finished_reason="abort",model_name="/job/model/Qwen2.5-0.5B-Instruct"} 0.0
# HELP http_request_size_bytes Content length of incoming requests by handler. Only value of header is respected. Otherwise ignored. No percentile calculated.
# TYPE http_request_size_bytes summary
http_request_size_bytes_count{handler="/v1/completions"} 2.0
http_request_size_bytes_sum{handler="/v1/completions"} 312.0
http_request_size_bytes_count{handler="/v1/chat/completions"} 1.0
http_request_size_bytes_sum{handler="/v1/chat/completions"} 268.0"""

    def load_example_metrics(cls):
        script_path = os.path.abspath(__file__)
        script_dir = os.path.dirname(script_path)
        data_path = os.path.join(script_dir, "metrics_example.txt")
        with open(data_path, 'r') as f:
            return f.read().strip()

    def test_parse_metrics_text(cls):
        example_metrics = cls.load_example_metrics()
        m = Metrics()
        result = m.parse_metric_text(example_metrics)
        assert result is not None

    def mock_get_available_instances(self, role: PDRole):
        return self.all_ins[role]

    def mock_requests_get_without_abnormal(self, *args, **kwargs):
        port = urlparse(args[0]).port
        return MockResponse({"metrics": self.metrics_template}, 200)

    def mock_requests_get_with_abnormal(self, *args, **kwargs):
        port = urlparse(args[0]).port
        if port in [8000, 8001, 8002, 8003]:
            return MockResponse({"metrics": self.metrics_template}, 200)
        return MockResponse(None, 404)

    @patch('motor.coordinator.core.instance_manager.InstanceManager.get_available_instances')
    def test_get_available_instances(self, mock_get_available_instances):
        mock_get_available_instances.side_effect = self.mock_get_available_instances

        assert InstanceManager().get_available_instances(PDRole.ROLE_P) == self.all_ins[PDRole.ROLE_P]
        assert InstanceManager().get_available_instances(PDRole.ROLE_D) == self.all_ins[PDRole.ROLE_D]
        assert InstanceManager().get_available_instances(PDRole.ROLE_U) == self.all_ins[PDRole.ROLE_U]

    @patch('requests.get')
    def test_requests_get(self, mock_requests_get):
        mock_requests_get.side_effect = self.mock_requests_get_without_abnormal

        for port in [8000, 8001, 8002, 8003, 8004, 8005]:
            assert requests.get(f"http://localhost:{port}/metrics").status_code == 200

        mock_requests_get.side_effect = self.mock_requests_get_with_abnormal

        for port in [8000, 8001, 8002, 8003]:
            assert requests.get(f"http://localhost:{port}/metrics").status_code == 200
        for port in [8004, 8005]:
            assert requests.get(f"http://localhost:{port}/metrics").status_code == 404

    @patch('motor.coordinator.core.instance_manager.InstanceManager.get_available_instances')
    @patch('requests.get')
    def test_prometheus_metrics_handler(self, mock_requests_get, mock_get_available_instances):
        mock_get_available_instances.side_effect = self.mock_get_available_instances
        mock_requests_get.side_effect = self.mock_requests_get_without_abnormal

        metric_listener = MetricsListener()
        metric_listener.update_metrics()
        result = metric_listener.prometheus_metrics_handler()
        assert result is not None

    @patch('motor.coordinator.core.instance_manager.InstanceManager.get_available_instances')
    @patch('requests.get')
    def test_prometheus_instance_metrics_handler(self, mock_requests_get, mock_get_available_instances):
        mock_get_available_instances.side_effect = self.mock_get_available_instances
        mock_requests_get.side_effect = self.mock_requests_get_without_abnormal

        metric_listener = MetricsListener()
        metric_listener.update_metrics()
        result = metric_listener.prometheus_instance_metrics_handler()
        assert result is not None

    @patch('motor.coordinator.core.instance_manager.InstanceManager.get_available_instances')
    @patch('requests.get')
    def test_prometheus_metrics_handler_abnormal(self, mock_requests_get, mock_get_available_instances):
        mock_get_available_instances.side_effect = self.mock_get_available_instances
        mock_requests_get.side_effect = self.mock_requests_get_with_abnormal # abnormal

        metric_listener = MetricsListener()
        metric_listener.update_metrics()
        result = metric_listener.prometheus_metrics_handler()
        assert result is not None

    @patch('motor.coordinator.core.instance_manager.InstanceManager.get_available_instances')
    @patch('requests.get')
    def test_prometheus_instance_metrics_handler_abnormal(self, mock_requests_get, mock_get_available_instances):
        mock_get_available_instances.side_effect = self.mock_get_available_instances
        mock_requests_get.side_effect = self.mock_requests_get_with_abnormal # abnormal

        metric_listener = MetricsListener()
        metric_listener.update_metrics()
        result = metric_listener.prometheus_instance_metrics_handler()
        assert result is not None

